import { AxiosError, AxiosResponse } from "axios";
import * as _ from "lodash-es";
import * as LosslessJson from "lossless-json";
import { EMPTY, type Observable, ObservableInput, ObservedValueOf, OperatorFunction, TimeoutError, UnaryFunction, catchError, count, defaultIfEmpty, defer, delayWhen, expand, filter, finalize, from, ignoreElements, lastValueFrom, map, merge, mergeMap, of, pipe, retry, tap, timeout, timer, toArray } from "rxjs";

export class RxjsPipe
{
    /**
     * 计算流一共发送了多少数据：先计算好，再往下走
     * @param cb cb 计算结果回调函数
     * @returns 
     */
    static countThanDo<T>(cb: (count: number) => unknown): UnaryFunction<Observable<T>,Observable<T>>
    {
        return pipe(
            toArray(),
            tap(list => cb(list.length)),
            mergeMap(list => from(list))
        )
    }

    /**
     * 计算流一共发送了多少数据：先往下走，最后再计算
     * @param cb 计算结果回调函数
     * @returns 
     */
    static countWithDo<T>(cb: (count: number) => unknown): UnaryFunction<Observable<T>,Observable<T>>
    {
        let counter = 0
        return pipe(
            tap(() => counter++),
            finalize(() => {cb(counter);counter = 0})
        )
    }

    /**
     * 重试前后执行回调函数，回调时传入错误
     * @param delay 延迟多少毫秒
     * @param cb 错误回调函数
     * @param count 重试次数
     * @returns 
     */
    static retryAfterDo<T>(delay: number, cb: (err: any) => unknown, count?:number): UnaryFunction<Observable<T>,Observable<T>>
    {
        return pipe(
            catchError(err => {cb(err);throw err}),
            retry({delay, count})
        )
    }

    /**
     * 超时并自动重试
     * @param timeoutNum 超时时间设置
     * @param delay 重试延迟时间
     * @param cb 重试前回调函数
     * @param count 最大重试次数
     * @returns 
     */
    static timeoutRetryAfterDo<T>(timeoutNum: number, delay: number, cb?: (err: TimeoutError<any, any>) => unknown, count?:number): UnaryFunction<Observable<T>,Observable<T>>
    {
        let currentCount = 0
        return pipe(
            timeout(timeoutNum),
            catchError((err,baseObs) =>
            {
                if(!_.isNil(count) && currentCount > count) throw err

                if(err instanceof TimeoutError)
                {
                    if(cb) cb(err)
                    return merge(
                        timer(delay).pipe(ignoreElements(),finalize(() => currentCount++)),
                        baseObs,
                        1
                    )
                }
                throw err
            })
        )
    }

    /**
     * 捕捉错误并返回EMPTY
     * @param cb 错误回调函数
     * @returns 
     */
    static catchErrorReturnEMPTY<T>(cb: (err: any) => unknown):UnaryFunction<Observable<T>,Observable<T>>
    {
        return pipe(
            catchError(err => {cb(err);return EMPTY}),
        )
    }

    /**
     * 无限重试axiosError
     * 
     * 一般都是网络问题，而网络问题应该在一段时间内恢复，所以无限重试是合理的
     * @param retryDelay 重试等待时间
     * @param axiosErrorCallback 重试前回调，方便log
     * @returns 
     */
    static catchAxiosError<T extends AxiosResponse<any,any>>(
        retryDelay: number = 30e3,
        axiosErrorCallback?: (err:AxiosError) => void
    ): OperatorFunction<T,T>
    {
        return pipe(
            catchError((err, baseObs) =>{
                if(err instanceof AxiosError)
                {
                    if(axiosErrorCallback) axiosErrorCallback(err)
                    return timer(retryDelay).pipe(mergeMap(() => baseObs))
                }
                throw err
            })
        )
    }

    /**
     * 将字符串JSON数据的大数字类型转换为string类型并最终返回一个对象
     */
    static bigNumberObjToString(): UnaryFunction<Observable<string>,Observable<any>>
    {
        return pipe(
            map(v => (<any>LosslessJson.parse(v, undefined, (numlike) => LosslessJson.isSafeNumber(numlike)? Number(numlike):numlike))),
        )
    }

    /**
     * 将axios返回的text的字符串JSON数据的大数字类型转换为string类型并最终返回一个对象
     * @returns 
     */
    static axiosBigNumberObjToString<T extends AxiosResponse>(): OperatorFunction<T,T>
    {
        return pipe(
             mergeMap(v => of(<string>v.data).pipe(
                 RxjsPipe.bigNumberObjToString(),
                 map(newData => ({...v,data: newData}))
             )),
        )
    }

    /**
     * 非操作符--递归地对任意对象中的null进行删除
     * @param dataTodo 要处理的对象
     * @param handlePipe 允许进行最后修改的回调函数
     * @returns 返回 处理对象的流
     */
    static deleteNullFromObjObs<T extends any>(
        dataTodo: T,
        handlePipe: UnaryFunction<Observable<IBoxForDeleteNullFromObjObs>,Observable<IBoxForDeleteNullFromObjObs>> = pipe()
    ):Observable<T>
    {
        let returnObs = <any>{}
        return of(<IBoxForDeleteNullFromObjObs>{path:"returnObs", data:dataTodo}).pipe(
            expand(box =>
            {
                if(_.isObjectLike(box.data))
                {
                    let boxList = <IBoxForDeleteNullFromObjObs[]>[]
                    _.forEach(box.data, (v,k) =>
                    {
                        let bb = {path:`${box.path}.${k}`, data:v}
                        boxList.push(bb)
                    })
                    return from(boxList)
                }
                return EMPTY
            }),
            filter(box => !_.isNull(box.data) && !_.isObjectLike(box.data)),
            handlePipe,
            tap(box =>
            {
                _.set(returnObs, box.path,box.data)
            }),
            count(),
            map(() => <T>returnObs["returnObs"])
        )
    }

    /**
     * 单线程mergeMap并控制内存回收
     * 
     * 通过设置一个内存触发值，当内存占用超过这个值的时候，下一次循环就延迟1ms
     * 
     * 这样保证CPU时间来GC，当然CPU不一定会马上GC，所以在成功GC前会一直触发
     * @param getTagetObs 类似mergeMap的参数，需要返回一个流
     * @param memoryTrigger 设置内存触发数值
     * @param getLoopInfo 可选，每轮循环的当前次数和内存值
     * @param getTriggerInfo 可选，触发内存回收时每轮循环的当前次数和内存值
     * @returns 
     */
    static mergeMapSingleThreadObs<T, R>(
        getTagetObs: (value1: T) => Observable<R>,
        memoryTrigger: number = 500,
        getLoopInfo?: (loopIndex: number, memoryAddCount: number) => void,
        getTriggerInfo?: (loopIndex: number, memoryAddCount: number) => void
    ): OperatorFunction<T, R>
    {
        /**
         * 内存开始值
         */
        let memoryUsageStart = 0
        /**
         * 内存增长监控，单位是MB
         */
        let memoryAddCount = 0

        let i = 0
        return pipe(
            /**
             * 设置内存开始值
             */
            tap(() =>
            {
                if(memoryUsageStart == 0) memoryUsageStart = process.memoryUsage.rss()
            }),
            mergeMap(value1 => getTagetObs(value1).pipe(
                /**
                 * 内存占用超500MB时，中断1ms时间让cpu有空GC
                 */
                delayWhen(() =>
                {
                    getLoopInfo ? getLoopInfo(i, memoryAddCount) : 0
                    
                    if(memoryAddCount > memoryTrigger)
                    {
                        getTriggerInfo ? getTriggerInfo(i, memoryAddCount) : 0
                        return timer(1)
                    }
                    return of(1)
                }),
                /**
                 * 计算记录内存占用情况
                 */
                finalize(() =>
                {
                    i++
                    memoryAddCount = (process.memoryUsage.rss() - memoryUsageStart) / 1024 / 1024
                }),
            ),1)
        )
    }

    /**
     * 操作符--mergeMap一个流
     * @param tagetObs 
     * @param eachPipe 
     * @param threadCount1 
     * @param threadCount2 
     * @returns 
     */

    static mergeMapEachInObs<T, R, O extends ObservableInput<any>>(
        tagetObs: Observable<R>,
        eachPipe: (value1: T, value2: R) => O,
        threadCount1?: number,
        threadCount2?: number
    ): OperatorFunction<T, ObservedValueOf<O>>
    {
        return RxjsPipe.eachInObs(
            tagetObs,
            (v) => pipe(
                mergeMap(v2 => eachPipe(v, v2),threadCount2)
            ),
            threadCount1
        )
    }

    /**
     * 操作符--foreach一个流
     * @param tagetObs 
     * @param eachPipe 
     * @param threadCount1 
     * @returns 
     */
    static eachInObs<T, R, O>(
        tagetObs: Observable<R>,
        eachPipe: (value1: T) => OperatorFunction<R, R>,
        threadCount1?: number
    ): OperatorFunction<T, R>
    static eachInObs<T, R, O>(
        tagetObs: Observable<R>,
        eachPipe: (value1: T) => OperatorFunction<R, O>,
        threadCount1?: number
    ): OperatorFunction<T, O>
    static eachInObs<T, R, O>(
        tagetObs: Observable<R>,
        eachPipe: (value1: T) => OperatorFunction<R, O|R>,
        threadCount1?: number
    ): OperatorFunction<T, O|R>
    {
        return pipe(
            mergeMap(v => tagetObs.pipe(
                eachPipe(v)
            ),threadCount1)
        )
    }

    /**
     * 在一个流调用前运行一个函数
     * @param func 
     * @param obs 
     * @returns 
     */
    static runBefore<T>(func: () => unknown, obs: Observable<T>): Observable<T>
    {
        return defer(() =>
        {
            func()
            return obs
        })
    }

    /**
     * 在一个流运行前后分布调用两个函数
     * @param beforeFunc 
     * @param obs 
     * @param afterFunc 
     * @returns 
     */
    static runBeforeAndAfter<T>(
        beforeFunc: () => unknown,
        obs: Observable<T>,
        afterFunc: () => unknown
    ): Observable<T>
    {
        return RxjsPipe.runBefore(beforeFunc, obs).pipe(finalize(afterFunc))
    }

    /**
     * 等待一个流完成
     * 
     * 代替lastValueFrom不需要处理空值报错
     * @param obs 流
     * @returns 无返回值
     */
    static async awaitObs<T>(obs: Observable<T>): Promise<void>
    {
        await lastValueFrom(obs.pipe(defaultIfEmpty(1)))
    }
}

export interface IBoxForDeleteNullFromObjObs {path:string, data:any}
